package com.gpj.learn.rocketmq.base;

import com.gpj.learn.rocketmq.CommonConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * @className SimpleConsumer
 * @description
 *   基本的消息消费者,用于其他模式下,消费者不需要特殊处理时使用
 * @author GPJ
 * @date 2021/4/22 21:58
 * @version 1.0
 **/
@Slf4j
public class BaseConsumer {
	public static void main(String[] args) throws Exception {
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("BasePushConsumerGroup");
		consumer.setNamesrvAddr(CommonConstants.ROCKETMQ_NAMESERVER_ADDR);
		consumer.subscribe("TestTopic", "*");
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
		consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
			for (MessageExt message : list) {
				log.info("接收到消息:{}", message);

			}
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		});
		consumer.start();
		log.info("consumer starting...");
	}
}
